package jp.ne.hatena.d.digitalsoul.warikan.repositories.impl

import java.util.ArrayList
import java.util.Date
import java.util.HashMap
import java.util.UUID

import scala.collection.JavaConversions._

import org.apache.log4j.Logger
import org.apache.cassandra.thrift._
import ConsistencyLevel._

import org.apache.thrift.protocol.TBinaryProtocol
import org.apache.thrift.protocol.TProtocol
import org.apache.thrift.transport.TSocket
import org.apache.thrift.transport.TTransport

import jp.ne.hatena.d.digitalsoul.warikan.models._
import jp.ne.hatena.d.digitalsoul.warikan.cassandra._
import CassandraMutation._
import CassandraRange._

class PartyRepositoryImpl {

	private val logger:Logger = Logger.getLogger(classOf[PartyRepositoryImpl])

	def nextId:String = UUID.randomUUID.toString
	
	def add(party:Party) = {
		
		logger.info("party added. partyId:" + party.id)
		
		val timestamp = System.currentTimeMillis
		
		// inner map
		val innerMap = new HashMap[String, java.util.List[Mutation]] // CFName, Column
		
		// party
		val partyColumns = new ArrayList[Mutation]
		
		val partyNameMutation = createColumnMutation("partyName".getBytes("utf-8"), party.name.getBytes("utf-8"), timestamp)
		partyColumns.add(partyNameMutation)
		
		val scheduleMutation = createColumnMutation("schedule".getBytes("utf-8"), java.lang.Long.toString(party.schedule.getTime).getBytes("utf-8"), timestamp)
		partyColumns.add(scheduleMutation)

		val locationMutation = createColumnMutation("location".getBytes("utf-8"), party.location.getBytes("utf-8"), timestamp)
		partyColumns.add(locationMutation)
		
		val sumMutation = createColumnMutation("sum".getBytes("utf-8"), party.sum.toString.getBytes("utf-8"), timestamp)
		partyColumns.add(sumMutation)
		
		innerMap.put("Party", partyColumns)
		
		// participants
		val participantRows = new ArrayList[Mutation]
		party.participants.foreach { it =>
			
			val participant = it._2
						
			val participantId:String = UUID.randomUUID.toString
			
			val participantColumns = new ArrayList[Column]
			val userNameColumn = new Column("userName".getBytes("utf-8"), participant.name.getBytes("utf-8"), timestamp)
			participantColumns.add(userNameColumn)
			val roleNameColumn = new Column("roleName".getBytes("utf-8"), participant.role.getBytes("utf-8"), timestamp)
			participantColumns.add(roleNameColumn)
			
			val participantMutation = createSuperColumnMutation(participantId.getBytes("utf-8"), participantColumns)
			participantRows.add(participantMutation)
			
		}
		innerMap.put("Participant", participantRows)
		
		// allots
		val allotRows = new ArrayList[Mutation]
		party.allots.foreach { it =>
			
			val allot = it._2
			
			val allotId:String = UUID.randomUUID.toString
			
			val allotColumns = new ArrayList[Column]
			val userNameColumn = new Column("userName".getBytes("utf-8"), allot.name.getBytes("utf-8"), timestamp)
			allotColumns.add(userNameColumn)
			val amountColumn = new Column("amount".getBytes("utf-8"), allot.amount.toString.getBytes("utf-8"), timestamp)
			allotColumns.add(amountColumn)
			
			val allotMutation = createSuperColumnMutation(allotId.getBytes("utf-8"), allotColumns)
			allotRows.add(allotMutation)
			
		}
		innerMap.put("Allot", allotRows)
		
		// mutation map
		val mutationMap = new HashMap[String, java.util.Map[String, java.util.List[Mutation]]] // key, innerMap
		mutationMap.put(party.id, innerMap)
		
		// insert
		val transport:TTransport = new TSocket("localhost", 9160)
		val protocol:TProtocol = new TBinaryProtocol(transport)
		val client:Cassandra.Client = new Cassandra.Client(protocol)
		transport.open			
			
		try {
			client.batch_mutate("warikan", mutationMap, ONE)
		} finally {
			transport.close
		}
		
	}
	
	
	
	def forPartyId(id:String):Option[Party] = {
		
		val transport:TTransport = new TSocket("localhost", 9160)
		val protocol:TProtocol = new TBinaryProtocol(transport)
		val client:Cassandra.Client = new Cassandra.Client(protocol)
		transport.open
		
		try {
			
			//party
			val partyColumnParent = new ColumnParent("Party")
			val partyCoSs:java.util.List[ColumnOrSuperColumn] = client.get_slice("warikan", id, partyColumnParent, sliceRangeAll, ONE)
			logger.info("partyCoSs:" + partyCoSs)
			
			if (partyCoSs.isEmpty) return None
			
			val party:Party = convertToParty(id, partyCoSs)
			
			// participants
			val participantColumnParent = new ColumnParent("Participant")
			
			val participantCoSs:java.util.List[ColumnOrSuperColumn] = client.get_slice("warikan", id, participantColumnParent, sliceRangeAll, ONE)
			logger.info("participantCoSs:" + participantCoSs)

			participantCoSs.foreach { participantCoS =>
				val participant = convertToParticipant(participantCoS)
				party.participants += participant.name -> participant
			}
						
			// allots
			val allotColumnParent = new ColumnParent("Allot")
								
			val allotCoSs:java.util.List[ColumnOrSuperColumn] = client.get_slice("warikan", id, allotColumnParent, sliceRangeAll, ONE)
			logger.info("allotCoSs:" + allotCoSs)
						
			allotCoSs.foreach { allotCoS =>
				val allot = convertToAllot(allotCoS)
				party.allots += allot.name -> allot
			}
			
			return Option(party)
			
		} finally {
			transport.close
		}

	}
	
	def forAll:List[Party] = {
		
		val transport:TTransport = new TSocket("localhost", 9160)
		val protocol:TProtocol = new TBinaryProtocol(transport)
		val client:Cassandra.Client = new Cassandra.Client(protocol)
		transport.open
		
		try {
			
			// party
			val partyColumnParent = new ColumnParent("Party")
			val partyKeySlices:java.util.List[KeySlice] = client.get_range_slices("warikan", partyColumnParent, sliceRangeAll, keyRangeAll, ONE)
		
			if (partyKeySlices isEmpty) return List[Party]()
			
			// party keys
			val partyKeys = new ArrayList[String]()
			partyKeySlices.foreach { partyKeySlice =>
				partyKeys.add(partyKeySlice.key)
			}
			
			// participant
			val participantColumnParent = new ColumnParent("Participant")
			val participantMap:java.util.Map[String, java.util.List[ColumnOrSuperColumn]] = client.multiget_slice("warikan", partyKeys, participantColumnParent, sliceRangeAll, ONE)
			
			// allot
			val allotColumnParent = new ColumnParent("Allot")
			val allotMap:java.util.Map[String, java.util.List[ColumnOrSuperColumn]] = client.multiget_slice("warikan", partyKeys, allotColumnParent, sliceRangeAll, ONE)
			
			// convert
			var partyList = List[Party]()
			partyKeySlices.foreach { partyKeySlice =>

				logger.debug("partyKeySlice:" + partyKeySlice)
				
				// party
				val party = convertToParty(partyKeySlice.key, partyKeySlice)
				
				// participant
				participantMap(party.id).foreach { participantCoS =>
					val participant = convertToParticipant(participantCoS)
					party.participants += participant.name -> participant
				}
				
				// allot
				allotMap(party.id).foreach { allotCoS =>
					val allot = convertToAllot(allotCoS)
					party.allots += allot.name -> allot
				}
				
				partyList ::= party
			}
			
			return partyList
			
		} finally {
			transport.close
		}
		
	}
	
	def paymentOf(id:String, name:String):Int = {
		assert(id != null)
		assert(name != null)
		
		val transport:TTransport = new TSocket("localhost", 9160)
		val protocol:TProtocol = new TBinaryProtocol(transport)
		val client:Cassandra.Client = new Cassandra.Client(protocol)
		transport.open
		
		try {
		
			val allotColumnParent = new ColumnParent("Allot")
			val allotCoSs:java.util.List[ColumnOrSuperColumn] = client.get_slice("warikan", id, allotColumnParent, sliceRangeAll, ONE)
			
			for (allotCoS <- allotCoSs) {
				val allot = convertToAllot(allotCoS)
				if (allot.name == name) return allot.amount 
			}
			
			return 0

		} finally {
			transport.close
		}
	}
	
	
	private def convertToParty(id:String, partyCoSs:java.util.List[ColumnOrSuperColumn]):Party = {
		val party:Party = new Party(id)
		partyCoSs.foreach { cos =>
			populateParty(party, cos.column)
		}
		party
	}
	
	private def convertToParty(id:String, partyKeySlice:KeySlice):Party = {
		val party:Party = new Party(id)
		val partyColumns = partyKeySlice.columns
		partyColumns.foreach { cos =>
			populateParty(party, cos.column)
		}
		party
	}
	
	private def populateParty(party:Party, column:Column) = {
		new String(column.name, "utf-8") match {
			case "partyName" => party.name  = new String(column.value, "utf-8")
			case "schedule" => party.schedule  = new Date(new String(column.value, "utf-8").toLong)
			case "location" => party.location = new String(column.value, "utf-8")
			case "sum" => party.sum = new String(column.value, "utf-8").toInt
		}
	}
	
	private def convertToParticipant(participantCoS:ColumnOrSuperColumn):Participant = {
		val columns = participantCoS.super_column.columns
		var userName:String = null
		var roleName:String = null
		columns.foreach { column =>
			new String(column.name, "utf-8") match {
				case "userName" => userName = new String(column.value, "utf-8")
				case "roleName" => roleName = new String(column.value, "utf-8")
			}
		}
		new Participant(userName, roleName)
	}
	
	private def convertToAllot(allotCoS:ColumnOrSuperColumn):Allot = {
		val columns = allotCoS.super_column.columns
		var userName:String = null
		var amount:Int = 0
		columns.foreach { column =>
			new String(column.name, "utf-8") match {
				case "userName" => userName = new String(column.value, "utf-8")
				case "amount" => amount = new String(column.value, "utf-8").toInt
			}
		}
		new Allot(userName, amount)
	}

}